package cn.doitedu.rule_marketing_platform.controller;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.Jedis;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.*;

@RestController
public class RuleController {

    Connection connection;
    Connection dorisConnection;
    Jedis jedis;

    public RuleController() throws SQLException {

        connection = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit47", "root", "root");
        dorisConnection = DriverManager.getConnection("jdbc:mysql://doitedu:9030/dwd?useUnicode=true&useSSL=false", "root", "root");

        jedis = new Jedis("doitedu", 6379);

    }


    @RequestMapping("/abc")
    public String addRule(@RequestBody String data) throws SQLException, IOException {

        JSONObject jsonObject = JSON.parseObject(data);
        String rule_id = jsonObject.getString("rule_id");
        String model_id = jsonObject.getString("model_id");


        // 查询模型信息
        String sql = "select * from rule_model_info where model_id = ?";
        PreparedStatement ps = connection.prepareStatement(sql);
        ps.setString(1, model_id);
        ResultSet rs = ps.executeQuery();
        rs.next();

        // 模型是否需要预圈选人群
        int isNeedPreselect = rs.getInt("is_need_preselect");
        // 模型是否需要预统计历史条件
        int isNeedCrossStatistic = rs.getInt("is_need_cross_statistic");
        // 模型对应的运算机源代码
        String calculatorCode = rs.getString("calculator_code");


        String rule_desc = "";

        int status = 1;

        String rule_author = "深似海的男人";

        Timestamp create_time = new Timestamp(System.currentTimeMillis());
        Timestamp modify_time = new Timestamp(System.currentTimeMillis());


        // 做人群预圈选
        // 将圈选的人群放入bitmap
        Roaring64Bitmap preSelectUsersBitmap = new Roaring64Bitmap();
        byte[] preSelectUsers = null;
        if (isNeedPreselect == 1) {
            for(long i = 1 ; i< 10000L ; i++) {
                preSelectUsersBitmap.add(i);
            }
        }
        // 序列化bitmap成为字节数组
        ByteArrayOutputStream bout = new ByteArrayOutputStream();
        DataOutputStream dout = new DataOutputStream(bout);
        preSelectUsersBitmap.serialize(dout);

        preSelectUsers = bout.toByteArray();

        /* *
         * 历史段数据预统计：
         */
        if (isNeedCrossStatistic == 1) {
            // 从规则参数中，取出 历史数据的统计条件
            JSONObject cross = jsonObject.getJSONObject("cross_history_static_condition");
            String eventId = cross.getString("event_id");
            Long startTime = cross.getLong("start_time");
            Long endTime = System.currentTimeMillis();
            // 把这个统计截止点，补充到规则参数中去
            jsonObject.put("cross_statistic_end", endTime);

            String querySql = "select user_id,count(1) as cnt from dwd.user_actions where event_id = ? and action_time between ? and ?  group by user_id";

            PreparedStatement pst1 = dorisConnection.prepareStatement(querySql);
            pst1.setString(1, eventId);
            pst1.setLong(2, startTime);
            pst1.setLong(3, endTime);

            ResultSet resultSet = pst1.executeQuery();
            while (resultSet.next()) {
                long userId = resultSet.getLong("user_id");
                long cnt = resultSet.getLong("cnt");

                // 如果这个用户属于预圈选人群，就把他的跨历史数据统计条件的历史值，发布到redis
                if ( (isNeedPreselect == 1 &&  preSelectUsersBitmap.contains(userId) )  ||   isNeedPreselect == 0 ) {
                    jedis.hset(rule_id, userId + "", cnt + "");
                }
            }
        }


        /**
         * 把规则元数据 信息，写入规则元数据表
         */
        PreparedStatement pst = connection.prepareStatement("insert into rule_meta values(?,?,?,?,?,?,?,?,?,?)");
        pst.setString(1, rule_id);
        pst.setString(2, rule_desc);
        pst.setString(3, model_id);
        pst.setString(4, jsonObject.toJSONString());

        pst.setBytes(5, preSelectUsers);

        pst.setInt(6, status);

        pst.setString(7,calculatorCode);

        pst.setString(8, rule_author);
        pst.setTimestamp(9, create_time);
        pst.setTimestamp(10, modify_time);

        pst.execute();

        System.out.println(data);
        return "ok";
    }


}
